# -*- coding: utf-8 -*- 
"""
@Author : Chan ZiWen
@Date : 2022/6/17 09:33
File Description:
Columns: [
    gdl_zdfw_sc_bluetooth_device.log_id,
    gdl_zdfw_sc_bluetooth_device.tv_mac,
    gdl_zdfw_sc_bluetooth_device.active_id,
    gdl_zdfw_sc_bluetooth_device.bluetooth_name,
    gdl_zdfw_sc_bluetooth_device.mac,
    gdl_zdfw_sc_bluetooth_device.trade_name,
    gdl_zdfw_sc_bluetooth_device.device_name,
    gdl_zdfw_sc_bluetooth_device.create_time,
    gdl_zdfw_sc_bluetooth_device.rssi,
    gdl_zdfw_sc_bluetooth_device.distance,
    gdl_zdfw_sc_bluetooth_device.log_time,
    gdl_zdfw_sc_bluetooth_device.day
    ]
log_id,tv_mac,active_id,mac,rssi,distance,log_time,day,sum(mac)
存入流程：

        1， 获取到每天计算好的结果，
        2， 从hive结果表中读取每个15天的分析结果
        3，
mode tv_mac active_id bluetooth_name mac current_time datetime final_time log_time

"""
import os
import time
import yaml
import json
import requests
import argparse
import vthread
import pandas as pd
from pyhive import hive

from analysis_utils import sleep, wake
from datetime import datetime, timedelta

work_dir = os.path.dirname(__file__)
args = yaml.safe_load(open(os.path.join(work_dir, "lifetime.yaml")))

config = args['config_hive']
interval_15 = args['interval_15']

# 用于并行处理数据
pool1 = vthread.pool(40, gqueue=1)  # 开40个伺服线程，组名为1
# 用于并行处理数据
pool2 = vthread.pool(40, gqueue=2)  # 开40个伺服线程，组名为2

sleep_id_mac_time = []      # List[dict]
wake_id_mac_time = []       # List[dict]
# 记录15天最终时间的数据
final_time_info = []


def Connect(configs):
    con = hive.Connection(
                host=configs['host'],
                port=configs['port'],
                auth='LDAP',
                username=configs['user'],
                password=configs['password'],
                database=configs['database'])
    return con


def Batch_select(date: str = None, mode: str = 'wake'):
    """
        Get all the pair of (active_id, mac) (mobile devices) according day and log_time (20:00-02:00 or 03:00-10:00)
    by "group by"
    """
    con = Connect(config)
    cursor = con.cursor()
    # 预设分析时间
    UpStartTime = " 03:00:00"
    UpEndTime = " 10:00:00"
    DownStartTime = " 20:00:00"
    DownEndTime = " 02:00:00"

    start = time.time()

    if mode == 'wake':
        # 生成时间戳
        startTime = date + UpStartTime
        endTime = date + UpEndTime
        stamp = int(time.mktime(time.strptime(startTime, "%Y-%m-%d %H:%M:%S")) * 1000)
        etamp = int(time.mktime(time.strptime(endTime, "%Y-%m-%d %H:%M:%S")) * 1000)
        # 执行查询
        """,collect_set(rssi),collect_set(distance)"""
        cursor.execute(
            f'select tv_mac, active_id, mac, collect_list(log_time), collect_list(distance), count(*) '
            f'from {config["table"]} '
            f'where log_time>="{stamp}" AND log_time<="{etamp}" '
            f'group by tv_mac, active_id, bluetooth_name, mac '
            f'having count(*) > 200 ')
    else:
        # 生成时间戳
        startTime = date + DownStartTime
        endTime = date + DownEndTime
        stamp = int(time.mktime(time.strptime(startTime, "%Y-%m-%d %H:%M:%S")) * 1000)
        etamp = int((time.mktime(time.strptime(endTime, "%Y-%m-%d %H:%M:%S")) + 86400) * 1000)    # 加上一天的秒数
        # 执行查询
        cursor.execute(
            f'select tv_mac, active_id, mac, collect_list(log_time), collect_list(distance), count(*) '
            f'from {config["table"]} '
            f'where log_time>="{stamp}" AND log_time<="{etamp}" '
            f'group by tv_mac, active_id, bluetooth_name, mac '
            f'having count(*) > 200 ')
    y = cursor.fetchall()
    cursor.close()
    con.close()
    print("Duration(get all source data) time: ", time.time() - start)
    print(f"All device's data length: {len(y)} ({mode})")
    return y


@pool1
def single_fn(tv_mac: str = None, active_id: str = None, mac: str = None, log_time: str = None, distance: str = None,
              mode: str = 'wake', date: str = None):
    """
    :param log_time:  shape (n,)
    :param distance:  shape (n,)
    :return:
   {"mode":1,"activeId":"34","mac":"3243","tvMac":"23423","oneDayTime":15013452,"final15dTime":"12:00","analysisDatetime":"2022-02-12"},
    """
    # str to list
    log_time = log_time.strip(']').strip('[').split(',')
    distance = distance.strip(']').strip('[').split(',')

    # sort by string
    log_time, distance = zip(*sorted(zip(log_time, distance), key=lambda s: (s[0], s[1])))
    if mode == 'sleep':
        ans = sleep(log_time, distance)
        if ans is not None:
            ans_date = int(ans) // 1000
            # ans_date = datetime.fromtimestamp(ans_date).strftime("%Y-%m-%d %H:%M:%S")
            # sleep_id_mac_time.append([2, tv_mac, active_id, mac, ans_date, date])
            # to dict
            sleep_id_mac_time.append(
                {
                    "mode": 2,
                    "tvMac": tv_mac,
                    "activeId": active_id,
                    "mac": mac,
                    "oneDayTime": ans_date,
                    "final15dTime": "",
                    "analysisDatetime": date
                })
        else:
            ans_date = None
    elif mode == 'wake':
        ans = wake(log_time, distance)
        if ans is not None:
            ans_date = int(ans) // 1000
            # ans_date = datetime.fromtimestamp(ans_date).strftime("%Y-%m-%d %H:%M:%S")
            # wake_id_mac_time.append([1, tv_mac, active_id, mac, ans_date, date])
            wake_id_mac_time.append(
                {
                    "mode": 1,
                    "tvMac": tv_mac,
                    "activeId": active_id,
                    "mac": mac,
                    "oneDayTime": ans_date,
                    "final15dTime": "",
                    "analysisDatetime": date
                })
        else:
            ans_date = None
    else:
        raise Warning(f"Can't find this mode ({mode}), there are some choices ('sleep', 'wake')")

    # print(f"mode: {mode} - active_id: {active_id} - mac: {mac} - ans: {ans} - ans: {ans_date}")


def batch_analysis(date: str = None, mode: str = None):
    """ multi-thread processing
    procedure:
        1, creating threads
        2, ordering data by log_time
        3, calling single mac analysis function
    """
    res_list = Batch_select(date, mode)
    # date_stamp = int(time.mktime(time.strptime(date, "%Y-%m-%d")))
    for res in res_list:
        tv_mac, active_id, mac, log_time, distance = res[0], res[1], res[2], res[3], res[4]
        single_fn(tv_mac, active_id, mac, log_time, distance, mode, date)       # timestamp


def post(date, datas, url, headers, mode='sleep'):
    datas = json.dumps(datas)
    response = json.loads(requests.post(url, data=datas, headers=headers).text)
    if response['code'] != 1000:
        raise RuntimeError(f"[{mode}] {date}({response})")


def save2ck(date):
    """
    [
    {"mode":1,"activeId":"34","mac":"3243","tvMac":"23423","oneDayTime":15013452,"final15dTime":"12:00","analysisDatetime":"2022-02-12"},
    {"mode":1,"activeId":"34","mac":"3243","tvMac":"23423","oneDayTime":15013452,"final15dTime":"12:00","analysisDatetime":"2022-02-12"},
    ]
    :return:
    """
    url = args["url1_inner"]
    headers = {"Content-Type": "application/json"}
    # if len(wake_id_mac_time) > 0:
    #     response = json.loads(requests.post(url, data=json.dumps(wake_id_mac_time), headers=headers).text)
    #     if response['code'] != 1000:
    #         raise RuntimeError(f"[wake] {date}({response})")
    # if len(sleep_id_mac_time) > 0:
    #     response = json.loads(requests.post(url, data=json.dumps(sleep_id_mac_time), headers=headers).text)
    #     if response['code'] != 1000:
    #         raise RuntimeError(f"[sleep] {date}({response})")
    inter_10000 = 10000
    n = len(wake_id_mac_time)
    if n > 0:
        # 超过1w条：则分批存
        if n > inter_10000:
            nb = n // inter_10000
            for i in range(nb):
                datas_part = wake_id_mac_time[i * inter_10000: (i + 1) * inter_10000]
                post(date, datas_part, url, headers, 'wake')
            if len(wake_id_mac_time[(i + 1) * inter_10000:]) > 0:
                post(date, wake_id_mac_time[(i + 1) * inter_10000:], url, headers, 'wake')
        else:
            post(date, wake_id_mac_time, url, headers, 'wake')
    n = len(sleep_id_mac_time)
    if n > 0:
        # 超过1w条：则分批存
        if n > inter_10000:
            nb = n // inter_10000
            for i in range(nb):
                datas_part = sleep_id_mac_time[i * inter_10000: (i + 1) * inter_10000]
                post(date, datas_part, url, headers)
            if len(sleep_id_mac_time[(i + 1) * inter_10000:]) > 0:
                post(date, sleep_id_mac_time[(i + 1) * inter_10000:], url, headers)
        else:
            post(date, sleep_id_mac_time, url, headers)


@pool2
def util_15d(df: pd.DataFrame = None, Info: list = None):
    #
    mode = Info[0]
    active_id = Info[2]
    mac = Info[4]
    cur_time = Info[5]
    if len(df.values) == 0:
        final_time = cur_time
    else:
        ans = df.loc[df['mode'] == mode and df['active_id'] == active_id and df['mac'] == mac]['current_time']  # 取出 current_time 这一列
        res_days = ans.values

        # 开始计算最终值
        if len(res_days) <= 0:
            final_time = cur_time
        elif len(res_days) == 1:
            final_time = (res_days[0] + cur_time) // 2
        else:
            ans = ans.append([{'current_time': cur_time}], ignore_index=True)
            # remove maximum and minimum
            ans.sort_values(inplace=True)
            final_time = ans.iloc[1:-1].mean()
    log_time = int(time.time())
    Info.append(final_time)
    Info.append(log_time)
    final_time_info.append(Info)


def main(date):
    """
    Duration time:  515.8442580699921
    data length:  7219

    Duration time:  111.48466920852661
    data length:  7219
    """
    modes = ['sleep', 'wake']
    start = time.time()
    # res = [('91787602', 'B00247B99B3C', '[1653513595185,1653508612025,1653507472585,1653506510107,1653516774023,1653517197820,1653512752028,1653525299801,1653522600690,1653510112434,1653525714462,1653514010897,1653513662140,1653513413003,1653520733469,1653518336734,1653515571485,1653516600929,1653511365871,1653522894875,1653520133826,1653513950884,1653508320641,1653505376794,1653524092486,1653519541779,1653516054806,1653512164806,1653517969414,1653518553873,1653513120223,1653512457278,1653510717238,1653507595496,1653515873535,1653512520218,1653523617773,1653507082263,1653523019913,1653510535469,1653506031946,1653505858974,1653521270631,1653515427487,1653513240646,1653516417095,1653507002535,1653520787803,1653512631836,1653524393923,1653524813324,1653521035216,1653519837907,1653510947220,1653514971905,1653508019913,1653524631047,1653511671518,1653522837831,1653512275798,1653526199886,1653523551411,1653526078552,1653524514134,1653512870156,1653523169315,1653508071775,1653525833667,1653527758953,1653517139093,1653506819091,1653525416287,1653518815340,1653505615383,1653521990421,1653516534492,1653507532699,1653509933297,1653518040597,1653525050583,1653516898845,1653515275722,1653511975410,1653508555935,1653520949452,1653517766975,1653523313397,1653525598664,1653526018715,1653512699030,1653511252012,1653526448660,1653525470849,1653512990660,1653507834920,1653511612732,1653509153776,1653506149735,1653520374064,1653505499216,1653517434483,1653516111042,1653517262384,1653508668278,1653516293752,1653518311582,1653516373795,1653511909116,1653507181232,1653513709391,1653520559546,1653513470713,1653518458371,1653515222310,1653508975759,1653509379002,1653523730403,1653519893168,1653515940163,1653519414836,1653519718942,1653508368654,1653519296364,1653519773307,1653513305151,1653508500761,1653515998817,1653522111856,1653506457474,1653523251551,1653518104520,1653510594473,1653507348290,1653523185555,1653507409353,1653505988246,1653524284570,1653509693292,1653518873953,1653517916117,1653525185381,1653521395680,1653517316274,1653512338317,1653508189330,1653509878522,1653525935041,1653523371734,1653506277103,1653516471300,1653520618310,1653524938993,1653516964436,1653511795473,1653523837159,1653518686746,1653506691508,1653512930224,1653510891105,1653518388650,1653522960396,1653509090077,1653527992597,1653524767441,1653521517777,1653524159935,1653510363107,1653522415528,1653518988418,1653520972025,1653506932395,1653512577465,1653509639733,1653506876121,1653522182618,1653520435438,1653522085782,1653505253884,1653522296261,1653520499167,1653522664178,1653526259115,1653520677398,1653508250182,1653524337046,1653522234239,1653515763770,1653521343022,1653520028582,1653514131794,1653526376062,1653525362820,1653522529856,1653525648708,1653524875410,1653521456422,1653519182985,1653519959980,1653516239991,1653511554398,1653521579922,1653514854677,1653510477999,1653514680769,1653515515872,1653514546930,1653512034041,1653510769313,1653516655998,1653510238014,1653506098262,1653524578013,1653519355758,1653524993419,1653513363230,1653510177990,1653516721370,1653515154388,1653507239483,1653511487921,1653526558005,1653510291044,1653505739039,1653519116539,1653505794187,1653514504859,1653505672723,1653524216523,1653521101800,1653521158020,1653520187683,1653513893659,1653509215287,1653507776762,1653508731271,1653523679281,1653522768969,1653519599438,1653515452393,1653511131904,1653525538332,1653520084393,1653514373481,1653514318309,1653513056603,1653526676946,1653519052581,1653517111703,1653512219805,1653511018677,1653507717400,1653526500449,1653522483814,1653520321145,1653528046413,1653509448459,1653528289864,1653508786168,1653509750047,1653520854561,1653511855263,1653524705831,1653514073478,1653514915607,1653510833751,1653509301787,1653516838531,1653518163683,1653523107370,1653526854999,1653518756251,1653517009794,1653519232951,1653507118612,1653506571770,1653517494636,1653521641076,1653517553708,1653514199274,1653518649188,1653514742150,1653526731695,1653523978337,1653518936497,1653505307301,1653509604917,1653522745076,1653508914245,1653521931682,1653511430381,1653525775988,1653517852353,1653512809444,1653508438204,1653528190733,1653509034156,1653516171994,1653515689891,1653517710128,1653525239619,1653521209478,1653517377069,1653514619388,1653510416680,1653507956514,1653506392474,1653525119395,1653513840770,1653526128386,1653519671150,1653515061484,1653515637104,1653506217977,1653521873175,1653524029015,1653521813613,1653509511252,1653515087460,1653507898887,1653524456262,1653523923984,1653511191852,1653506751327,1653522356727,1653512103385,1653511323880,1653517621402,1653518216907,1653505551584,1653521692116,1653514825681,1653506638315,1653512396436,1653509812963,1653508131832,1653519481278,1653510050927,1653506343553,1653511732133,1653510662963,1653509989907,1653505914949]', '[8.99,8.99,11.94,10.36,10.36,8.99,10.36,10.36,8.99,8.99,11.94,8.99,8.99,8.99,8.99,8.99,8.99,8.99,8.99,8.99,8.99,10.36,10.36,8.99,10.36,10.36,10.36,11.94,10.36,8.99,8.99,8.99,10.36,8.99,10.36,10.36,10.36,8.99,11.94,8.99,8.99,8.99,10.36,8.99,10.36,8.99,10.36,8.99,10.36,11.94,10.36,8.99,10.36,11.94,8.99,10.36,8.99,8.99,10.36,8.99,8.99,8.99,10.36,8.99,10.36,8.99,8.99,10.36,11.94,8.99,8.99,10.36,8.99,10.36,10.36,10.36,8.99,8.99,10.36,10.36,8.99,7.8,8.99,8.99,10.36,8.99,10.36,8.99,10.36,8.99,8.99,10.36,8.99,8.99,8.99,7.8,10.36,8.99,8.99,8.99,10.36,10.36,10.36,8.99,8.99,10.36,10.36,10.36,8.99,8.99,10.36,8.99,8.99,10.36,10.36,8.99,8.99,8.99,10.36,8.99,8.99,8.99,10.36,11.94,8.99,8.99,10.36,8.99,10.36,8.99,8.99,10.36,8.99,10.36,8.99,8.99,10.36,8.99,8.99,10.36,11.94,10.36,8.99,8.99,8.99,10.36,10.36,10.36,8.99,8.99,8.99,11.94,10.36,8.99,11.94,10.36,8.99,8.99,8.99,8.99,8.99,10.36,10.36,8.99,8.99,10.36,8.99,11.94,10.36,8.99,7.8,8.99,8.99,8.99,10.36,10.36,10.36,8.99,8.99,10.36,10.36,10.36,8.99,8.99,10.36,8.99,10.36,8.99,8.99,10.36,10.36,10.36,10.36,10.36,8.99,8.99,8.99,10.36,10.36,8.99,8.99,8.99,8.99,8.99,8.99,10.36,8.99,8.99,8.99,10.36,8.99,10.36,8.99,11.94,10.36,8.99,8.99,8.99,10.36,8.99,11.94,10.36,8.99,8.99,10.36,8.99,8.99,10.36,8.99,10.36,10.36,8.99,11.94,10.36,8.99,10.36,8.99,8.99,8.99,8.99,10.36,8.99,8.99,8.99,8.99,11.94,10.36,8.99,8.99,8.99,10.36,10.36,8.99,8.99,11.94,8.99,11.94,8.99,8.99,10.36,10.36,8.99,10.36,10.36,8.99,10.36,10.36,8.99,8.99,10.36,8.99,8.99,8.99,8.99,10.36,8.99,10.36,10.36,10.36,8.99,8.99,11.94,8.99,8.99,8.99,8.99,11.94,8.99,10.36,10.36,11.94,8.99,8.99,8.99,11.94,8.99,10.36,8.99,10.36,10.36,10.36,8.99,10.36,8.99,10.36,10.36,10.36,10.36,8.99,10.36,10.36,8.99,8.99,8.99,10.36,10.36,10.36,10.36,8.99,10.36,8.99,10.36,8.99,8.99,10.36,8.99,8.99,10.36,8.99,10.36,10.36,10.36,8.99,8.99,8.99,10.36,10.36,8.99,8.99,8.99,10.36,10.36]', 342)]
    times = []
    for mode in modes:
        tmp = time.time()
        batch_analysis(date, mode)
        times.append(time.time() - tmp)
    vthread.pool.wait(gqueue=1)
    print(f"Total duration(read & analysis) time: {(time.time() - start) / 60}(m), wake duration time: {times[0] / 60}(m), "
          f"sleep duration time: {times[1] / 60}(m), ")
    # 添加将list内容存储到mysql的函数，通过调用接口

    # save2ck(date)
    wake_id_mac_time.clear()
    sleep_id_mac_time.clear()


def Parsers():
    parser = argparse.ArgumentParser("For the wake and sleep analysis the parser")
    parser.add_argument('-d', '--date', default=None, type=str)
    return parser.parse_args()


if __name__ == '__main__':
    parsers = Parsers()
    date = parsers.date
    if date is None:
        date = datetime.now().date() - timedelta(days=1)
    print(f"Begin of the ({date})")
    main(date)

    # date_end = "2022-06-30"
    # date_stamp = int(time.mktime(time.strptime(date, "%Y-%m-%d")))
    # date_stamp_end = int(time.mktime(time.strptime(date_end, "%Y-%m-%d")))
    # oneday = 86400
    # length = (date_stamp_end - date_stamp) // oneday
    # for _ in range(length):
    #     date_stamp += oneday
    #     date_new = datetime.fromtimestamp(date_stamp).strftime("%Y-%m-%d")
    #     print(f"Begin of the ({date_new})")
    #     main(date_new)


